xds: External Processor Server Interceptor - #12889

Allow unit test to pass CacheChannelManager for in-process channel.

Fix mock ext-proc service to handle all phases of the request-response events to avoid test hanging.

The test failed with "too many messages" error due to sending an empty body message to the data plane server during the "half-close" phase . For a unary RPC, this was interpreted as a second request message, which is invalid. I've updated handleRequestBodyResponse and onExternalBody to only call super.sendMessage() or super.onMessage() if the body content is non-empty. This prevents the redundant empty message from being sent to the data plane while still allowing the external processor to signal the end of the stream.

The stream between the filter and the external processor was never being closed on the client side, causing the InProcessChannel and InProcessServer to hang during shutdown while waiting for the active RPC to terminate.
To fix this, I have updated ExternalProcessorFilter.java to ensure the control plane stream is gracefully closed when the data plane RPC completes or is cancelled.
Changes made:
1. Closing on Completion: In ExtProcClientCall.onNext, once the ResponseTrailers handshake is finished and the application has been notified via proceedWithClose(), I now call extProcClientCallRequestObserver.onCompleted().
2. Handling Cancellation: I overridden the cancel() method in ExtProcClientCall. If the data plane RPC is cancelled by the application, the filter now also cancels the external processor stream with an error, ensuring all resources are freed.
3. Observability Mode Fix: In observability mode, since we don't wait for a ResponseTrailers message from the server, I added logic to ExtProcListener.onClose() to close the external processor stream immediately after sending the final trailers.
These changes ensure proper lifecycle management of the side-channel RPC.

…Shares backpressure logic with observability mode.

…companying message ever with half-close.

…by overriding ClientCall.requestMessages(int) in ExtProcClientCall. Also introduce null check for extProcClientCallRequestObserver in isReady since it may be called on the call even before start is called that initializes it.

…ation of the External Processor filter coordinates data across the application thread, the data plane response thread, and the external processor's response thread. To ensure thread safety and compliance with the gRPC contract, the

  following synchronization measures were implemented:

  1. Thread-Unsafe StreamObserver
   * Challenge: The gRPC StreamObserver used to send messages to the external processor is not thread-safe. Concurrent calls to its onNext(), onCompleted(), and onError() methods from different threads can corrupt the internal state of the communication
     channel. Additionally, calling isReady() on the observer while another thread is sending data can lead to race conditions.
   * Fix: All interactions with the external processor's StreamObserver—including data transmission (onNext), terminal signals (onCompleted, onError), and readiness checks (isReady)—are now protected by the lock object.

  2. ClientCall.Listener Serialization Contract
   * Challenge: gRPC requires that all callbacks to an application's ClientCall.Listener (such as onHeaders, onMessage, and onReady) be strictly serialized. Because these events can be triggered by either the backend server or the external processor,
     there was a risk of overlapping callbacks.
   * Fix: The logic that delivers events to the application's Listener is now synchronized using the lock. This ensures that even if multiple threads attempt to "unblock" and deliver buffered metadata or status simultaneously, the application receives
     them in a single, non-overlapping sequence.

  3. Visibility and Consistency of Internal State
   * Challenge: The filter maintains several internal state variables, such as buffers for response metadata and flags to track the lifecycle of the call. If these are accessed concurrently without synchronization, one thread might act on stale data,
     potentially leading to duplicate headers or incorrect flow control decisions.
   * Fix: Access to all internal state and control flags is now guarded by the lock. Furthermore, the flag indicating whether request headers have been processed was marked as volatile to ensure its state is immediately visible across threads during
     high-frequency checks like sendMessage().

  4. Synchronizing Terminal Signals
   * Challenge: Closing or faulting the external processor's stream while another thread is still attempting to send data can cause crashes or undefined behavior in the gRPC transport.
   * Fix: All terminal signals (onCompleted and onError) sent to the external processor's StreamObserver are synchronized with the same lock used for sending data. This ensures that the stream is only terminated after any ongoing data transfers have
     safely finished.

Fix some incorrect handlings done using buffered messages, there should be no need to buffer messages except in the case of observability mode when headers have been not yet been sent.

Fix missing coordinated synchronizations between threads.

…rove concurrency in the External Processor filter is complete. The implementation now employs a more granular three-lock strategy:

   1. streamLock: Guards all interactions with the extProcClientCallRequestObserver. This ensures the gRPC StreamObserver to the external processor is never accessed concurrently, protecting its internal state.
   2. requestLock: Manages the outbound flow control. It guards the headersSent flag and the pendingActions queue, coordinating the transition from the initial buffering phase to active delivery to the backend server.
   3. responseLock: Serializes all callbacks to the application's Listener (onHeaders, onMessage, onClose, onReady). It also guards shared response state like savedHeaders and savedStatus. This ensures strict compliance with the gRPC contract while
      fixing a potential race condition in onExternalBody.

  By decoupling the request and response data planes, the filter now supports full-duplex concurrency where outbound messages do not block inbound server responses. All lock acquisitions were carefully refactored to be sequential, maintaining a
  consistent order and guaranteeing deadlock-free execution.

…tName (grpc#12644)"

We want to synchronize the behavior across all gRPC languages, and also with envoy.

This reverts commit 0ef1b39.

This change is a no-op. The create() form is clearer than positional
arguments to a heavily overloaded constructor.

Guard this behavior change behind the RFC 3986 parser flag.

Google Play Services needs min Android API level of 23 (Android 6.0
Marshmallow).

Fixes [grpc#11474](grpc#11474).

The backoff timer is only used when serializeRetries=true, and that
exists to match the old/current pick_first's behavior as closely as
possible. InternalSubchannel.updateAddresses() would take no action when
in TRANSIENT_FAILURE; it would update the addresses and just wait for
the backoff timer to expire.

Note that this only impacts serializeRetries=true; in the other cases we
do want to start trying to the new addresses immediately, because the
backoff timers are in the subchannels.

Note that this change was also important because requestConnection() can
be directly triggered by the user with channel.getState(true), and that
shouldn't defeat the backoff timer.

Since we're only supporting API levels 23+, all the supported Android
versions handle multidex natively, and without any bugs to workaround.

Also bump some minSdkVersion that didn't get updated in fa7b52b so
that multiDex is actually enabled by default.

See also b/476359563

DnsNameResolver discards refresh requests if it has been too soon after
the last refresh, because the result is assumed to be identical to the
previous fetch. Android itself will adhere to the RR's TTL, so
requesting too frequently shouldn't have been causing too much I/O, but
it could be causing extra CPU usage. Having some lower limit will reduce
the number of useless address updates into the LB tree.

30 seconds is the same as regular Java and Go/C++ (which copied Java as
a "reasonable" value). Note that other languages _delay_ the refresh
instead of _discarding_ the refresh, but there's no reason why the
existing discard behavior would cause much problem on Android vs normal
Java. Chrome apparently uses 1 minute, so this really looks like it
shouldn't cause problems as long as AndroidChannelBuilder is being used.

The internal result was needed before 90d0fab allowed addresses to
fail yet still provide attributes and service config. Now the code can
just use the regular API.

This does cause a behavior change where TXT records are looked up even
if address lookup failed, however that's actually what we wanted to
allow in 90d0fab by adding the new API. Also, the TXT code was added
in 2017 and it's now 2026 yet it is still disabled, so it's unlikely to
matter soon.

…#12697)

4de4718 upgraded android-interop-testing to SDK version 23, but this
had previously been avoided because it triggered a Gradle or AGP bug.
The race happened to not trigger locally or for the PR's CI and the
change was merged. But the problem still was present, and the CI is
failing to build very frequently.

This works around the problem by explicitly adding a dependency from
mergeExtDexDebug. I didn't see any other mergeExtDex tasks created, in
particular mergeExtDexRelease. Hopefully we can remove this after
upgrading AGP or Gradle, but at least we can move forward with newer
Android API levels again.

f36defa upgraded error_prone_annotations and 4de4718 bumped
minSdkVersion for android-interop-testing.

…DS server by ref-counting

This PR implements reusing the gRPC xDS transport (and underlying gRPC
channel) to the same xDS server by ref-counting, which is already
implemented in gRPC C++
([link](https://github.com/grpc/grpc/blob/5a3a5d53145b94895610825e783a8896a61a3c73/src/core/xds/grpc/xds_transport_grpc.cc#L399-L414))
and gRPC Go
([link](https://github.com/grpc/grpc-go/blob/81c7924ec9f5f4a01c18b82c9d67691c1cd93bd5/internal/xds/clients/grpctransport/grpc_transport.go#L78-L120)).
This optimization is expected to reduce memory footprint of the xDS
management server and xDS enabled clients as channel establishment and
lifecycle management of the connection is expensive.

* Implemented a map to store `GrpcXdsTransport` instances keyed by the
`Bootstrapper.ServerInfo` and each `GrpcXdsTransport` has a ref count.
Note, the map cannot be simply keyed by the xDS server address as the
client could have different channel credentials to the same xDS server,
which should be counted as different transport instances.
* When `GrpcXdsTransportFactory.create()` is called, the existing
transport is reused if it already exists in the map and increment its
ref count, otherwise create a new transport, store it in the map, and
increment its ref count.
* When `GrpcXdsTransport.shutdown()` is called, its ref count is
decremented and the underlying gRPC channel is shut down when its ref
count reaches zero.
* Note this ref-counting of the `GrpcXdsTransport` is different and
orthogonal to the ref-counting of the xDS client keyed by the xDS server
target name to allow for xDS-based fallback per [gRFC
A71](https://github.com/grpc/proposal/blob/master/A71-xds-fallback.md).

Prod risk level: Low
* Reusing the underlying gRPC channel to the xDS server would not affect
the gRPC xDS (ADS/LRS) streams which would be multiplexed on the same
channel, however, this means new xDS (ADS/LRS) streams and RPCs may fail
due to hitting the limit of `MAX_CONCURRENT_STREAMS`.

Tested:
* Verified end-to-end with a xDS enabled gRPC Java client communicating
to multiple different gRPC backend servers behind *different targets*
using the xDS management server for name resolution and endpoint
discovery. Verified gRPC xDS transport creation, ref-counting, reuse,
shutdown, deletion from map when ref count is zero all worked as
expected.

Implementation details / context:
* Used `java.util.concurrent.ConcurrentHashMap` APIs `compute` and
`computeIfPresent` where the entire method invocation is performed
atomically to achieve a concurrent and thread-safe solution which
follows Java best practices.

Alternatives considered: 
* Write own synchronization logic with synchronized block and locks.
After discussion internally, it was preferred to use existing
concurrency libraries which is less error-prone and should offer better
performance.

…rpc#12703)

grpc#12700)

### Description
This PR updates the "Outgoing Flow Control" section in the Manual Flow
Control example's README.

The previous documentation incorrectly implied that calling `onNext()`
on a stream would block if the underlying Netty buffer was full, thereby
limiting the send rate. This PR clarifies that `onNext()` does *not*
block, but rather queues the messages in memory, which can ultimately
lead to an `OutOfMemoryError` if messages are sent too quickly.

The updated text correctly advises developers to use
`CallStreamObserver.isReady()` to prevent this memory exhaustion, rather
than to avoid blocking.

Fixes grpc#12657

---------

Co-authored-by: Kannan J <kannanjgithub@google.com>

…pc#12718)

This alignment resolves a version skew issue that caused
NoClassDefFoundError crashes during instrumentation tests on Firebase
Test Lab.

Fixes
grpc#12703 (comment)

grpc#12705)

This PR addresses a race condition where ManagedChannelOrphanWrapper
could incorrectly log a "not shutdown properly" warning during garbage
collection when using directExecutor().

Changes:

Reference Management: Moved phantom.clearSafely() to execute after the
super.shutdown() calls to ensure the orphan tracker isn't detached
prematurely.

Reachability Fence: Added a reachability fence in shutdown() and
shutdownNow() to ensure the wrapper remains alive until the methods
return, preventing the JIT from marking it for early collection.

Regression Test: Added a test case that simulates a reference being held
on the stack to verify the fix and prevent future regressions.

Testing:
Verified with ./gradlew :grpc-core:test --tests
ManagedChannelOrphanWrapperTest -PskipAndroid=true.

Fixes grpc#12641

---------

Co-authored-by: Kannan J <kannanjgithub@google.com>

This adds triggerEvent/onEvent APIs to ServerCall and ServerCall.Listener,
routing them through ServerStream transport to ensure thread-safety
(especially for SerializeReentrantCallsDirectExecutor).

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

Refactors ExternalProcessorServerInterceptor to use the new custom events
framework (Hybrid Event Model), eliminating the syncContext lock and
serializing all callbacks (including ext_proc stub responses) on the
application executor. Also includes fixes for forwarding listeners to
propagate custom events and corrections to onReady notification timing.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

…pc#12979)

… behavior.

- Added unit tests in ServerImplTest for JumpToApplicationThreadServerStreamListener.triggerEvent.
- Added serverStream_triggerEvent_afterClose in AbstractTransportTest to verify events are ignored after stream closure.
- Updated Inbound.ServerInbound to check isClosed() before triggering events.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

…framework.

- Added unit tests in AbstractServerStreamTest for triggerEvent propagation and close behavior.
- Updated ContextsTest to cover onEvent propagation in ContextualizedServerCallListener.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

Wait for the server stream to be fully closed (via awaitClose) before
calling triggerEvent, to ensure the transport has processed the
cancellation and marked the listener as closed. This fixes flakiness in
slower transports like Jetty.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

- Consolidated locking under streamLock.
- Implemented queue-based flow control for request/response bodies and headers.

# Conflicts:
#	xds/src/main/java/io/grpc/xds/XdsServerWrapper.java

…from ext_proc were getting stuck in pendingMutatedRequestBodies because they were waiting for App demand to be drained. I fixed this by allowing end_of_stream_without_message to be drained immediately without requiring App demand.

… onCompleted due to client half-close) is handled correctly by the double-close protection and still propagates the original error status and not overwritten by Ok from the client half-close.

 Added metrics tests.

…olean.

ExternalProcessorServerInterceptor.java
:

1. Deferred Half-Close Propagation Bug (Null Delegate)
Issue: In proceedWithHalfClose(), if the delegate (app listener) was not set yet (null), the interceptor set requestSideClosed to true and returned. When the delegate was finally set via SetDelegateEvent, the subsequent call to proceedWithHalfClose() returned early because requestSideClosed was already true, preventing the half-close from ever reaching the application.
Fix: Modified proceedWithHalfClose() to return early without setting requestSideClosed if delegate is null, allowing the subsequent call to successfully propagate the half-close.
2. Observability Mode Close Hang Bug
Issue: In close(), the interceptor checked if there were outstanding response body requests. In observability mode, we return early from response processing, so outstandingResponseBodyRequests is never decremented. This caused close() to wait indefinitely for it to reach 0, hanging the call.
Fix: Bypassed the check for outstanding/pending messages in close() if observabilityMode is enabled.
3. Observability Mode Half-Close Propagation Bug
Issue: In onHalfClose(), if the mode was GRPC, we sent EOS to ext proc and returned without calling proceedWithHalfClose(). In observability mode, we don't process responses, so proceedWithHalfClose() was never called, and the app never saw the half-close.
Fix: Updated onHalfClose() to call proceedWithHalfClose() immediately if observabilityMode is enabled.

…ntrol

- Fix observability mode early close hang by proceeding with close immediately.
- Fix flow control close deferral to correctly check for pending/outstanding messages.
- Refine isRequestSideCompleted to only require half-close if request body is intercepted.
- Add Test 16 and Test 17 to verify these behaviors.
- Fix Test 18 regression by triggering failure on onCompleted.

TAG=agy
CONV=9ea901ca-a127-468c-a836-414f2154bf85

Implement Phase 4 of the coverage improvement plan.
- Add tests for observability mode early close with no trailers.
- Add tests for trailers-only close when headers are skipped (normal and observability modes).
- Improve flow control test to use manual flow control to trigger request buffering and deferred half-close coverage.

TAG=agy
CONV=9ea901ca-a127-468c-a836-414f2154bf85

Remove redundant IDLE check from isReady and redundant isExtProcStreamCompleted check from isSidecarReady.
Add tests and verify fallback behavior of isReady when stream is completed.

TAG=agy
CONV=9ea901ca-a127-468c-a836-414f2154bf85

TAG=agy
CONV=9ea901ca-a127-468c-a836-414f2154bf85

TAG=agy
CONV=9ea901ca-a127-468c-a836-414f2154bf85